package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.utils.JedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * @author gufg
 * @since 2025-09-22 10:00
 */
public class RedisSinkFunc extends RichSinkFunction<String> {
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = JedisUtil.getJedis();
    }

    @Override
    public void invoke(String jsonStr, Context context) throws Exception {
        // 转换
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        // 取key-name
        String orderId = jsonObj.getString("order_id");
        // 分数
        String time = jsonObj.getString("time");
        // value
        String value = jsonObj.getString("lng") + ":" + jsonObj.getString("lat");
        // 结束
        String driverId = jsonObj.getString("driver_id");


        // 保存数据到有序集合
        jedis.zadd(orderId, Double.parseDouble(time), value);

        // 设置Key过期24小时过期
        jedis.expire(orderId, 24 * 60 * 60L);

        // 订单结束
        if ("end".equals(driverId)) {
            // 实时订单集合
            jedis.srem("real_time_set", orderId);
            // 历史订单集合
            jedis.sadd("history_set", orderId);
        } else {
            // 实时订单集合
            jedis.sadd("real_time_set", orderId);
        }
    }

    @Override
    public void close() throws Exception {
        if (jedis != null) {
            jedis.close();
        }
    }
}
